package mq

import (
	"bytes"
	"fmt"
	"github.com/streadway/amqp"
)

type Callback func(msg string)

//rabbitmq 连接
func Connect() (*amqp.Connection, error) {
	conn, err := amqp.Dial("amqp://admin:admin@120.48.26.182:15672")
	return conn, err
}

//发送端函数 exchange=>交换机  queueName=>队列名称
func Publish(exchange string, queueName string, body string) error {
	//建立连接
	conn, err := Connect()
	if err != nil {
		return err
	}
	defer conn.Close()
	//创建通道
	channel, err := conn.Channel()
	if err != nil {
		return err
	}
	defer channel.Close()
	// 创建队列
	q, err := channel.QueueDeclare(queueName, false, false, false, false, nil)
	if err != nil {
		return err
	}
	//发送消息
	err = channel.Publish(exchange, q.Name, false, false, amqp.Publishing{
		//发布内容的类型
		ContentType: "text/plain",
		Body:        []byte(body),
	})
	return err
}

//接收者
func Consumer(exchange string, queueName string, callback Callback) {
	//建立连接
	conn, err := Connect()
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Close()
	//创建通道
	channel, err := conn.Channel()
	if err != nil {
		fmt.Println(err)
		return
	}
	defer channel.Close()
	// 创建队列
	q, err := channel.QueueDeclare(queueName, false, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
		return
	}
	//从队列中获取数据
	msgs, err := channel.Consume(q.Name, "", true, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
		return
	}
	forever := make(chan bool)
	go func() {
		//循环获取
		for d := range msgs {
			s := BytesToString(&d.Body)
			callback(*s)
		}
	}()
	fmt.Printf("waiting for message")
	<-forever
}

func BytesToString(b *[]byte) *string {
	s := bytes.NewBuffer(*b)
	r := s.String()
	return &r
}
